package com.tanghao.flink.demo.distinct;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.redisson.Redisson;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

public class UniqAnalysis extends RichFlatMapFunction<Tuple4<Long, String, String, String>, Tuple2<String, String>> {
    private String redisUrl = null;
    private String jobName = null;
    transient private RedissonClient redisson = null;
    transient private RBloomFilter<String> bloomFilter = null;
    private Tuple2<String, String> doaminFirstTime = new Tuple2<>("", "");
    @Override
    public void flatMap(Tuple4<Long, String, String, String> in, Collector<Tuple2<String, String>> out) {
        if (!this.bloomFilter.contains(in.f1 + in.f2 + in.f3)) {
            this.bloomFilter.add(in.f1 + in.f2 + in.f3);
            doaminFirstTime.f0 = String.valueOf(in.f0);
            doaminFirstTime.f1 = in.f1 +","+ in.f2 +"," + in.f3;
            out.collect(doaminFirstTime);
            //System.out.println("filter redisUrl:" + this.redisUrl + " bloomFilter:" + bloomFilter);
        }
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ExecutionConfig.GlobalJobParameters globalParams = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        Configuration globConf = (Configuration) globalParams;
        this.redisUrl = globConf.getString("redis_url", null);
        this.jobName = globConf.getString("job_name", null);
        //System.out.println("redisUrl:"+this.redisUrl);
        if(this.redisUrl!=null) {
            Config redissonConfig = new Config();
            redissonConfig.useSingleServer().setAddress(this.redisUrl);
            redisson = Redisson.create(redissonConfig);
            this.bloomFilter = redisson.getBloomFilter("first");
            this.bloomFilter.tryInit(100_000_000, 0.03);
            System.out.println(System.currentTimeMillis()+" ["+this.jobName+"] RichFilterFunction open new MessageSplitter bloomFilter:" + bloomFilter);
        }
    }

    @Override
    public void close() {
        System.out.println(System.currentTimeMillis()+" ["+this.jobName+"] redisson.shutdown()");
        this.redisson.shutdown();
    }
}
